package com.shujia.flink.async

import java.sql.Connection
import java.util.concurrent._
import java.util.function.{Consumer, Supplier}

import com.shujia.flink.async.Demo2Async.Score
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.async.{ResultFuture, RichAsyncFunction}

import scala.util.Random

object Demo2Async {

  def main(args: Array[String]): Unit = {

    /**
      * 异步连接mysql实现表关联
      *
      */


    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    val scores = env.socketTextStream("node1", 8888)
      .map(line => line.split(","))
      .filter(_.length == 3)
      .map(s => Score(s(0), s(1), s(2).toDouble))

    //异步io
    //orderedWait 返回结果有序
    //unorderedWait  先查询到先返回

    val joinDS = AsyncDataStream.unorderedWait(
      scores, //需要关联的流
      new AsyncDatabaseRequest(), //自定义异步方法
      10000, //么一个请求超时时间
      TimeUnit.MILLISECONDS, //
      100 //异步数量
    )


    joinDS.print()

    env.execute()
  }

  case class Score(stuId: String, couId: String, sco: Double) extends Serializable

}

class AsyncDatabaseRequest extends RichAsyncFunction[Score, String] {

  var threadPoll: ExecutorService = _

  //open 每一个分区执行一次，在taskmanager中执行
  override def open(parameters: Configuration): Unit = {
    //创建线程池， 每一个分区一个
    threadPoll = Executors.newFixedThreadPool(50)
  }

  //asyncInvoke 方法每一条数据被调用一次
  override def asyncInvoke(input: Score, resultFuture: ResultFuture[String]): Unit = {

    // 设置客户端完成请求后要执行的回调函数
    // 回调函数只是简单地把结果发给 future
    CompletableFuture.supplyAsync(new MySupplier(input), threadPoll)
      //回调方法，get返回之后执行
      .thenAccept(new Consumer[String] {
      override def accept(t: String): Unit = {
        //返回结果
        resultFuture.complete(Iterable(t))
      }
    })
  }

  //超时解决，不解决会报错
  override def timeout(input: Score, resultFuture: ResultFuture[String]): Unit = {
    println(" ==timeout==== " + input)
  }
}

class MySupplier(input: Score) extends Supplier[String] {
  //在后台启动一个线程去执行get方法
  override def get(): String = {

    //建立mysql连接
    val con = JDBCUtil.getConnection
    //通过学号查询学生信息
    val stat = con.prepareStatement("select * from student where id=?")
    stat.setString(1, input.stuId)

    //执行sql需要时间
    val resultSet = stat.executeQuery()

    var line = input.stuId + "\t" + input.couId + "\t" + input.sco

    //模型数据库访问延迟
//    val time = Random.nextInt(1000)
    //    Thread.sleep(time)
    //    println(input.stuId + " 查询数据库花费：" + time)


    if (resultSet.next()) {
      val name = resultSet.getString("name")
      val age = resultSet.getInt("age")
      val gender = resultSet.getString("gender")
      val clazz = resultSet.getString("clazz")

      line = line + "\t" + name + "\t" + age + "\t" + gender + "\t" + clazz
    }


    //关闭连接，将连接放回连接池
    con.close()

    line
  }
}


